import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/10/11 09:11
 */
public class Join_5 extends BaseSQLApp {
    public static void main(String[] args) {
        new Join_5().init(4001, 1, "Join_1");
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        /*tEnv.executeSql("create table ab4(" +
                            "id int, " +
                            "name string, " +
                            "age int " +
                            ")with(" +
                            " 'connector' = 'kafka', " +
                            " 'topic' = 'ab4', " +
                            " 'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                            " 'properties.group.id' = 'Join_5', " +
                            " 'scan.startup.mode' = 'latest-offset', " +
                            " 'format' = 'json')"
        );*/
    
        tEnv.executeSql("create table ab4(" +
                            " id int, " +
                            " name string, " +
                            " age int," +
                            " primary key(id) not enforced " +
                            ")with(" +
                            "  'connector' = 'upsert-kafka', " +
                            "  'topic' = 'ab4', " +
                            "  'properties.bootstrap.servers' = 'hadoop162:9092', " +
                            "  'key.format' = 'json', " +
                            "  'value.format' = 'json'" +
                            ")");
        
        tEnv.sqlQuery("select * from ab4").execute().print();
        
        
    }
}
/*
join:
    默认情况下, 会把数据永远存在状态中
    
left join
    如果右表先来, 则正常的每条数据都是新增
    
    如果左表先来, 会有一条结果是 左表有数据, 右表的字段用 null 补齐, 现在如果右表来了,在先删除,在再新增.以后再来都是新增
    
    对左表来说, ttl 表示的是 idle 的时间之后会清空数据
    
kafka连接器:
    普通的 kafka 只能写入新增的数据. 如果有更新和删除是无法写入的.
    
    左连接有删除(-d), 所以无法写入到普通的 kafka 中

upset kafka
    向upset kafka写数据的时候, 如果这次写的数据的 key 不存在, 就认为是新增
    如果 这次的 key 写入的时候 value 是 null, 就表示删除前面相同 key 的数据
    
    如果这次的 key 存在, value 不是 null, 表示是更新
    
  
注意:
    当用普通的 kafka 消费 topic 的时候, 如果value 是 null, 则自动忽略
    
    通过upset kafka从 topic 读取数据的时候, 则每次总是会从第一条数据开始构建整个过程. 一般写的时候用 upset kafka, 读的时候用普通的 kafka
   
总结消费规律:
    如果 kafka 的 topic 是由 upset-kafka 写入的
    
    用普通的 kafka 消费:
        不用管, null 会自动过滤
        所有的数据都认为是新增
    用 upset kafka消费:
        不用管 null,, 会自动识别删除
        每次都是从第一条数据开始构建整个过程
        (如果是从状态恢复, 就不会从头读了)
        
    用流的方式消费:
        需要自定义反序列化器.
        如果用 SimpleStringSchema 会出现空指针异常.
    
 */